package com.zz21

import com.gosun.{DataEncry, getResult}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import org.apache.spark.storage.StorageLevel

import scala.collection.JavaConversions.asScalaBuffer
import scala.collection.mutable.ArrayBuffer

object zz21_vw_rentalhouse {
  Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {
    val sparkSession = SparkSession
      .builder()
      .config("spark.network.timeout", "1200")
      .master("local[*]")
      .appName("SQLContextApp")
      .getOrCreate()
    val options = Map(
      "es.nodes.wan.only" -> "true",
      "es.nodes" -> "10.82.121.72",
      "es.port" -> "9200",
      "es.read.field.as.array.include" -> "arr1, arr2",
      "es.scroll.size" -> "10000",
      "es.input.use.sliced.partitions" -> "false"
    )
    val index = "zz21_vw_rentalhouse"
    val frame: DataFrame = sparkSession
      .read
      .format("es")
      .options(options)
      .load(index)
      .persist(StorageLevel.MEMORY_AND_DISK)
    //frame.show()

    val resSchema = StructType(
      List(
        StructField("TROUBLETYPE", StringType, true),
        StructField("LESSEEMOBILENUMBER", StringType, true),
        StructField("LESSEETELEPHONE", StringType, true),
        StructField("LESSEE", StringType, true),
        StructField("UPDATEUSER", StringType, true),
        StructField("UPDATEDATE", StringType, true),
        StructField("CREATEUSER", StringType, true),
        StructField("CREATEDATE", StringType, true),
        StructField("SOURCESSTATE", StringType, true),
        StructField("HOUSEID", StringType, true),
        StructField("LOGOUTREASON", StringType, true),
        StructField("LOGOUTDATE", StringType, true),
        StructField("ISEMPHASIS", StringType, true),
        StructField("UNDERLESSORPHONE", StringType, true),
        StructField("UNDERLESSORADDRESS", StringType, true),
        StructField("UNDERLESSORNAME", StringType, true),
        StructField("LESSEEIDCARD", StringType, true),
        StructField("ATTENTIONTYPE", StringType, true),
        StructField("ISATTENTION", StringType, true),
        StructField("CONDUITCOMPANYPHONE", StringType, true),
        StructField("CONDUITCOMPANYNAME", StringType, true),
        StructField("RENTALENDTIME", StringType, true),
        StructField("RENTALSTARTTIME", StringType, true),
        StructField("HASLEASSAGREEMENT", StringType, true),
        StructField("PAYTAXES", StringType, true),
        StructField("RENTALKIND", StringType, true),
        StructField("RENTALMETHOD", StringType, true),
        StructField("SIGNGUARANTEE", StringType, true),
        StructField("ISSIGNGUARANTEE", StringType, true),
        StructField("ISSAFETYCHANNEL", StringType, true),
        StructField("ISFIRECHANNELS", StringType, true),
        StructField("RENTMONTH", StringType, true),
        StructField("ROOMNUMBER", StringType, true),
        StructField("LESSORDATE", StringType, true),
        StructField("MANGERTYPES", StringType, true),
        StructField("HIDDENDANGERLEVEL", StringType, true),
        StructField("LESSORADDRESS", StringType, true),
        StructField("RENTALMOBILENUMBER", StringType, true),
        StructField("RENTALTELEPHONE", StringType, true),
        StructField("RENTALCERTIFICATENUMBE", StringType, true),
        StructField("RENTALCERTIFICATETYPE", StringType, true),
        StructField("RENTALPERSON", StringType, true),
        StructField("LESSORTYPE", StringType, true),
        StructField("HOUSEFILENUM", StringType, true),
        StructField("USAGE", StringType, true),
        StructField("ORGINTERNALCODE", StringType, true),
        StructField("ORGID", StringType, true),
        StructField("ID", StringType, true),
        StructField("LESSEEMOBILENUMBER_AES", StringType, true),
        StructField("LESSEETELEPHONE_AES", StringType, true),
        StructField("LESSEEIDCARD_AES", StringType, true),
        StructField("RENTALCERTIFICATENUMBE_AES", StringType, true)

      )
    )
    val value: RDD[Row] = frame.rdd.mapPartitions(iter => {
      val list = ArrayBuffer[Row]()
      while (iter.hasNext) {
        val row: Row = iter.next()
        val str1 = row.getAs[String]("TROUBLETYPE")
        val str2 = DataEncry.changPhone(row.getAs[String]("LESSEEMOBILENUMBER"))
        val str3 = DataEncry.changPhone(row.getAs[String]("LESSEETELEPHONE"))
        val str4 = DataEncry.changName(row.getAs[String]("LESSEE"))
        val str5 = DataEncry.changName(row.getAs[String]("UPDATEUSER"))
        val str6 = row.getAs[String]("UPDATEDATE")
        val str7 = DataEncry.changName(row.getAs[String]("CREATEUSER"))
        val str8 = row.getAs[String]("CREATEDATE")
        val str9 = row.getAs[String]("SOURCESSTATE")
        val str10 = row.getAs[String]("HOUSEID")
        val str11 = row.getAs[String]("LOGOUTREASON")
        val str12 = row.getAs[String]("LOGOUTDATE")
        val str13 = row.getAs[String]("ISEMPHASIS")
        val str14 = DataEncry.changPhone(row.getAs[String]("UNDERLESSORPHONE"))
        val str15 = DataEncry.changAddress(row.getAs[String]("UNDERLESSORADDRESS"))
        val str16 = DataEncry.changName(row.getAs[String]("UNDERLESSORNAME"))
        val str17 = DataEncry.changIDcard(row.getAs[String]("LESSEEIDCARD"))
        val str18 = row.getAs[String]("ATTENTIONTYPE")
        val str19 = row.getAs[String]("ISATTENTION")
        val str20 = DataEncry.changPhone(row.getAs[String]("CONDUITCOMPANYPHONE"))
        val str21 = row.getAs[String]("CONDUITCOMPANYNAME")
        val str22 = row.getAs[String]("RENTALENDTIME")
        val str23 = row.getAs[String]("RENTALSTARTTIME")
        val str24 = row.getAs[String]("HASLEASSAGREEMENT")
        val str25 = row.getAs[String]("PAYTAXES")
        val str26 = row.getAs[String]("RENTALKIND")
        val str27 = row.getAs[String]("RENTALMETHOD")
        val str28 = row.getAs[String]("SIGNGUARANTEE")
        val str29 = row.getAs[String]("ISSIGNGUARANTEE")
        val str30 = row.getAs[String]("ISSAFETYCHANNEL")
        val str31 = row.getAs[String]("ISFIRECHANNELS")
        val str32 = row.getAs[String]("RENTMONTH")
        val str33 = row.getAs[String]("ROOMNUMBER")
        val str34= row.getAs[String]("LESSORDATE")
        val str35 = row.getAs[String]("MANGERTYPES")
        val str36 = row.getAs[String]("HIDDENDANGERLEVEL")
        val str37 = DataEncry.changAddress(row.getAs[String]("LESSORADDRESS"))
        val str38 = DataEncry.changPhone(row.getAs[String]("RENTALMOBILENUMBER"))
        val str39 = DataEncry.changPhone(row.getAs[String]("RENTALTELEPHONE"))
        val str40 = DataEncry.changIDcard(row.getAs[String]("RENTALCERTIFICATENUMBE"))
        val str41 = row.getAs[String]("RENTALCERTIFICATETYPE")
        val str42 = row.getAs[String]("RENTALPERSON")
        val str43 = row.getAs[String]("LESSORTYPE")
        val str44 = row.getAs[String]("HOUSEFILENUM")
        val str45 = row.getAs[String]("USAGE")
        val str46 = row.getAs[String]("ORGINTERNALCODE")
        val str47 = row.getAs[String]("ORGID")
        val str48 = row.getAs[String]("ID")
        val str49 = DataEncry.changAES(row.getAs[String]("LESSEEMOBILENUMBER"))
        val str50 = DataEncry.changAES(row.getAs[String]("LESSEETELEPHONE"))
        val str51 = DataEncry.changAES(row.getAs[String]("LESSEEIDCARD"))
        val str52 = DataEncry.changAES(row.getAs[String]("RENTALCERTIFICATENUMBE"))
        val schema: GenericRowWithSchema = new GenericRowWithSchema(Array(
          str1,
          str2,
          str3,
          str4,
          str5,
          str6,
          str7,
          str8,
          str9,
          str10,
          str11,
          str12,
          str13,
          str14,
          str15,
          str16,
          str17,
          str18,
          str19,
          str20,
          str21,
          str22,
          str23,
          str24,
          str25,
          str26,
          str27,
          str28,
          str29,
          str30,
          str31,
          str32,
          str33,
          str34,
          str35,
          str36,
          str37,
          str38,
          str39,
          str40,
          str41,
          str42,
          str43,
          str44,
          str45,
          str46,
          str47,
          str48,
          str49,
          str50,
          str51,
          str52
        ), resSchema)
        list.append(schema)
      }
      list.iterator
    })
    val dataFrame: DataFrame = sparkSession.createDataFrame(value.coalesce(1).cache(), resSchema)
      .persist(StorageLevel.MEMORY_AND_DISK)
    val path = "C:\\Users\\Raichard\\Desktop\\宜春\\JSON\\zz21\\"
    val indespath = path + index + ".json"
    dataFrame.show(false)
    val str: String = dataFrame.toJSON.collectAsList.mkString("[", ",", "]")
    //执行方法
    println("执行方法")
    getResult.getData(str, indespath)
    /*dataFrame
      .write
      .mode(SaveMode.Overwrite)
      .format("json")
      .save(path)*/
    println("输出完成！")
    sparkSession.close()

  }

}
